package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.*;
import java.text.SimpleDateFormat;

public class Demo05MySQLSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Student> stuDS = env.addSource(new MyMySQLSource(5000));
        stuDS.print();

        env.execute();

    }
}

class Student {
    String id;
    String name;
    Integer age;
    String gender;
    String clazz;

    public Student(String id, String name, Integer age, String gender, String clazz) {
        this.id = id;
        this.name = name;
        this.age = age;
        this.gender = gender;
        this.clazz = clazz;
    }

    @Override
    public String toString() {
        return "Student{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", age=" + age +
                ", gender='" + gender + '\'' +
                ", clazz='" + clazz + '\'' +
                '}';
    }
}

class MyMySQLSource implements SourceFunction<Student> {
    Integer queryInterval;

    public MyMySQLSource(int queryInterval) {
        this.queryInterval = queryInterval;
    }

    @Override
    public void run(SourceContext<Student> sourceContext) throws Exception {
        // 建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata33", "root", "123456");

        // 创建Statement
//        Statement st = conn.createStatement();
        PreparedStatement pSt = conn.prepareStatement("select id,name,age,gender,clazz from students where last_mod >= ?");
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        // 查询
        while (true) {
            Date date = new Date(System.currentTimeMillis());
            String dateStr = dateFormat.format(date);
            pSt.setString(1,dateStr);
            ResultSet rs = pSt.executeQuery();
            while (rs.next()) {
                String id = rs.getString("id");
                String name = rs.getString("name");
                Integer age = rs.getInt("age");
                String gender = rs.getString("gender");
                String clazz = rs.getString("clazz");
                Student stu = new Student(id, name, age, gender, clazz);

                sourceContext.collect(stu);
            }

            Thread.sleep(this.queryInterval);
        }
    }

    @Override
    public void cancel() {

    }
}
